package batch;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import scala.Int;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class operator_join {
    public static void main(String[] args) throws Exception {
    //    创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    //    获取数据
        ArrayList<Tuple2<Integer,String>> data1 = new ArrayList<Tuple2<Integer,String>>();
        data1.add(new Tuple2<Integer, String>(1,"bing"));
        data1.add(new Tuple2<Integer, String>(2,"shui"));
        data1.add(new Tuple2<Integer, String>(3,"fan"));
        data1.add(new Tuple2<Integer, String>(4,"hong"));
        data1.add(new Tuple2<Integer, String>(2,"yang"));
        data1.add(new Tuple2<Integer, String>(3,"hong"));

        ArrayList<Tuple2<Integer,String>> data2 = new ArrayList<Tuple2<Integer,String>>();
        data2.add(new Tuple2<Integer, String>(1,"beijin"));
        data2.add(new Tuple2<Integer, String>(2,"shanghai"));
        data2.add(new Tuple2<Integer, String>(3,"wuhan"));
        data2.add(new Tuple2<Integer, String>(4,"gz"));


        DataSet<Tuple2<Integer,String>> user =  env.fromCollection(data1);
        DataSet<Tuple2<Integer,String>> city =  env.fromCollection(data2);

        System.out.println("*****join****");
        user.join(city).
                where(0).
                equalTo(0).
                with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple3<Integer, String,String>>() {
            public Tuple3<Integer, String,String> join(Tuple2<Integer, String> integerStringTuple2, Tuple2<Integer, String> integerStringTuple22) throws Exception {
                return new Tuple3<Integer,String,String>(integerStringTuple2.f0,integerStringTuple2.f1,integerStringTuple22.f1);
            }
        }).print();

        System.out.println("*****join****");

    }
}
